package com.example.demo.config;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.security.scram.ScramLoginModule;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;


@Slf4j
//@Configuration
public class KafkaConfig {

//    @Bean
//    public KafkaAdmin kafkaAdmin() {
//        Map<String, Object> configs = new HashMap<>();
//        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka.vggtest.com:9092");
//        // SSL 配置
//        configs.put("security.protocol", "SASL_SSL");
//        configs.put("ssl.truststore.location", "/jdk17/jdk17.0.12_7/lib/security/cacerts");
//        configs.put("ssl.truststore.password", "changeit");
//
//        // SASL 配置
//        configs.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
//        configs.put(SaslConfigs.SASL_JAAS_CONFIG,
//                ScramLoginModule.class.getName() + " required\n" +
//                        "username=\"admin\"\n" +
//                        "password=\"TestAdmin\";");
//
//        return new KafkaAdmin(configs);
//    }

//    @Bean
//    public NewTopic topic1() {
//        return createTopicIfNotExists("test-topic", 1, (short) 1);
//    }
//
//    @Bean
//    public NewTopic topic2() {
//        return createTopicIfNotExists("test-topic2", 1, (short) 1);
//    }

//    private NewTopic createTopicIfNotExists(String topicName, int partitions, short replicationFactor) {
//        try (AdminClient adminClient = AdminClient.create(kafkaAdmin().getConfigurationProperties())) {
//            // 检查主题是否已存在
//            ListTopicsResult listTopics = adminClient.listTopics();
//            Set<String> existingTopics = null;
//            try {
//                existingTopics = listTopics.names().get();
//            } catch (ExecutionException e) {
//                throw new RuntimeException(e);
//            }
//
//            for (String topic : existingTopics) {
//
//                log.info("exists topic:{}", topic);
//
//            }
//
//            if (!existingTopics.contains(topicName)) {
//                return new NewTopic(topicName, partitions, replicationFactor);
//            } else {
//                log.warn("主题：{} 已存在",topicName);
//                // 主题已存在，返回null避免重复创建
//                return null;
//            }
//        } catch (Exception e) {
//            throw new RuntimeException("Failed to check topic existence: " + topicName, e);
//        }
//    }



}